package com.exd.flink.sink.clickhouse;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions.Builder;
import org.apache.flink.streaming.api.datastream.DataStream;

import com.exd.flink.common.PropertiesTool;
import com.exd.flink.domain.ResourcesBean;
import com.exd.flink.domain.VDataBean;
import com.exd.flink.sink.business.ResourcesDataSink;
import com.exd.flink.sink.business.VdataFullIndexDataSink;

public class ClickHouseSink {
	
	/**
	 * 资源监控
	 */
	public static void inserResourcesData(DataStream<ResourcesBean> resourcesStream) {
		String sql = "INSERT INTO resource_monitoring (`vin`, `serviceId`, `serviceName`, `vComputeCpu`, `vComputeMem`, `vDataCpu`, `vDataMem`, `samplingTime`, `create_time`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?);";
		resourcesStream.addSink(JdbcSink.sink(sql,new ResourcesDataSink(),getJdbcExecution(),getJdbcConnection()));
	}
	
	/**
	 * VDCM全量数据索引
	 */
	public static void inserVdcmVdataPackageFullIndex(DataStream<VDataBean> fullDataIndexStream) {
		String sql = "INSERT INTO vdcm_vdata_package_full_index (`vin`, `serviceId`, `serviceName`, `deployId`, `collectTime`, `packetSequence`, `triggerStartTime`, `triggerEndTime`, `samplingTime`, `create_time`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);";
		fullDataIndexStream.addSink(JdbcSink.sink(sql,new VdataFullIndexDataSink(),getJdbcExecution(),getJdbcConnection()));
	}
	
	/**
	 * BDCM全量数据索引
	 */
	public static void inserBdcmVdataPackageFullIndex(DataStream<VDataBean> fullDataIndexStream) {
		String sql = "INSERT INTO bdcm_vdata_package_full_index (`vin`, `serviceId`, `serviceName`, `deployId`, `collectTime`, `packetSequence`, `triggerStartTime`, `triggerEndTime`, `samplingTime`, `create_time`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);";
		fullDataIndexStream.addSink(JdbcSink.sink(sql,new VdataFullIndexDataSink(),getJdbcExecution(),getJdbcConnection()));
	}
	
	/**
	 * ClickHouse批量配置
	 */
	public static JdbcExecutionOptions getJdbcExecution() {
		Builder builder = JdbcExecutionOptions.builder();
		//批量写入的数据大小
		builder.withBatchSize(10);
		//每次批量写入最小间隔时间
		builder.withBatchIntervalMs(1000);
		// 插入发生异常重试次数
		builder.withMaxRetries(3);
		return builder.build();
	}
	
	/**
	 * ClickHouse链接配置
	 */
	public static JdbcConnectionOptions getJdbcConnection() {
		JdbcConnectionOptions.JdbcConnectionOptionsBuilder jdbcConnectionOptionsBuilder = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder();
		ParameterTool parameterTool = PropertiesTool.getParameterTool();
		jdbcConnectionOptionsBuilder.withUrl(parameterTool.get("exd.clickhouse.url"));
		jdbcConnectionOptionsBuilder.withDriverName(parameterTool.get("exd.clickhouse.driver.name"));
//		jdbcConnectionOptionsBuilder.withUsername("root");
//		jdbcConnectionOptionsBuilder.withPassword("Exceed@2022!dev");
		return jdbcConnectionOptionsBuilder.build();
	}

}
